package com.chencong.online.function;

import com.chencong.online.bean.ItemViewCountBean;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author chencong
 * @Description 自定义窗口聚合函数
 * @Date 7:39 下午 2021/12/19
 * @Param
 **/
public class WindowItemCountResultFunc implements WindowFunction<Long, ItemViewCountBean, Tuple, TimeWindow> {
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<ItemViewCountBean> out) throws Exception {
        Long itemId = tuple.getField(0);
        long windowStart = window.getStart();
        long windowEnd = window.getEnd();
        Long count = input.iterator().next();//去除迭代器的值
        ItemViewCountBean itemViewCount = new ItemViewCountBean(
                itemId,
                windowStart,
                windowEnd,
                count
        );
        out.collect(itemViewCount);//收集器发出去
    }
}
